package com.example.job;

import com.example.latch.EtcdLatch;
import com.example.latch.EtcdLeaderLatch;
import com.example.latch.EtcdLeaderLatchListener;
import com.example.service.EtcdService;
import com.example.service.SysServerService;
import io.etcd.jetcd.*;
import io.etcd.jetcd.op.Op;
import io.etcd.jetcd.options.DeleteOption;
import io.etcd.jetcd.options.PutOption;
import io.etcd.jetcd.watch.WatchEvent;
import io.etcd.jetcd.watch.WatchResponse;
import io.fury.Fury;
import io.fury.config.Language;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/***
 * 定时任务的设置
 */
@Service
public class JobServiceImpl implements JobService, InitializingBean, DisposableBean {

    Logger logger = LoggerFactory.getLogger(JobServiceImpl.class);

    @Resource
    private EtcdService etcdService;

    @Resource
    private SysServerService sysServerService;

    private Fury fury;

    private EtcdLatch latch;

    private ExecutorService executorService;

    @Override
    public Fury getFury() {
        if (this.fury == null) {
            this.fury = Fury.builder().withLanguage(Language.JAVA).requireClassRegistration(true).build();
        }
        return fury;
    }

    @Override
    public Lease getLeaseClient() {
        return this.etcdService.getClient().getLeaseClient();
    }

    @Override
    public KV getKVClient() {
        return this.etcdService.getClient().getKVClient();
    }

    /***
     * 创建定时任务
     * @param jobId 任务id 唯一
     * @param jobClass 任务执行类
     * @param seconds 延迟时间 单位: 秒
     * @return
     */
    @Override
    public Boolean createAndSetJobInTx(String jobId, Class<? extends Job> jobClass, long seconds) throws Exception {
        JobInfo jobInfo = new JobInfo(jobId, jobClass.getName(), false, seconds);
        ByteSequence jobKey = ByteSequence.from(JOB_RUN + jobId, StandardCharsets.UTF_8);
        ByteSequence jobVal = ByteSequence.from(getFury().serialize(jobInfo));
        PutOption putOption = PutOption.newBuilder().withLeaseId(getLeaseClient().grant(seconds).get().getID()).build();
        return etcdService.getTxn().Then(Op.put(jobKey, jobVal, putOption)).commit().get().isSucceeded();
    }

    /***
     * 创建周期性的定时任务
     * @param jobId 任务id 唯一
     * @param jobClass 任务执行类
     * @param interval 执行间隔 单位: 秒
     * @return
     * @throws Exception
     */
    @Override
    public Boolean createAndSetRepeatJobInTx(String jobId, Class<? extends Job> jobClass, long interval) throws Exception {
        JobInfo jobInfo = new JobInfo(jobId, jobClass.getName(), true, interval);
        ByteSequence jobKey = ByteSequence.from(JOB_RUN + jobId, StandardCharsets.UTF_8);
        ByteSequence jobVal = ByteSequence.from(getFury().serialize(jobInfo));
        PutOption putOption = PutOption.newBuilder().withLeaseId(getLeaseClient().grant(interval).get().getID()).build();
        ByteSequence runJobKey = ByteSequence.from(JOB_LIST + jobId, StandardCharsets.UTF_8);
        return etcdService.getTxn().Then(Op.put(runJobKey, jobVal, PutOption.DEFAULT)).Then(Op.put(jobKey, jobVal, putOption)).commit().get().isSucceeded();
    }

    /***
     * 取消定时任务
     * @param jobId 任务id
     * @return
     */
    @Override
    public Boolean cancelJobInTx(String jobId) throws Exception {
        JobInfo jobInfo = new JobInfo(jobId, null, false, 0);
        ByteSequence jobKey = ByteSequence.from(JOB_RUN + jobId, StandardCharsets.UTF_8);
        ByteSequence jobVal = ByteSequence.from(getFury().serialize(jobInfo));
        ByteSequence runKey = ByteSequence.from(JOB_LIST + jobId, StandardCharsets.UTF_8);
        return etcdService.getTxn().Then(Op.put(jobKey, jobVal, PutOption.DEFAULT)).Then(Op.delete(runKey, DeleteOption.DEFAULT)).commit().get().isSucceeded();
    }


    /****
     * 更新定时任务
     * @param jobId
     * @param seconds
     * @return
     * @throws Exception
     */
    @Override
    public Boolean updateJobInTx(String jobId, long seconds) throws Exception {
        List<KeyValue> jobList = etcdService.getChildList(JOB_RUN + jobId);
        for (KeyValue keyValue : jobList) {
            JobInfo jobInfo = (JobInfo) getFury().deserialize(keyValue.getValue().getBytes());
            jobInfo.setInterval(seconds);
            this.createAndSetRepeatJobInTx(jobId, (Class<? extends Job>) Class.forName(jobInfo.getJobClass()), seconds);
        }
        return true;
    }

    /***
     * 初始化
     * @throws Exception
     */
    @Override
    public void afterPropertiesSet() throws Exception {
        this.getFury().register(JobInfo.class);
        this.executorService = Executors.newFixedThreadPool(20);
        this.executorService.execute(new JobInitThread());
        this.latch = new EtcdLeaderLatch(etcdService.getClient(), JOB_LEADER, sysServerService.getServerKey(), 3, new EtcdLeaderLatchListener() {
            @Override
            public void isLeader() {
                logger.warn("任务节点主节点: " + sysServerService.getServerKey() + " 开始工作 ...");
                etcdService.watchPrefKey(JOB_RUN, new Watch.Listener() {
                    @Override
                    public void onNext(WatchResponse response) {
                        List<WatchEvent> events = response.getEvents();
                        for (WatchEvent event : events) {
                            if (event.getEventType() == WatchEvent.EventType.DELETE) {
                                JobInfo jobInfo = (JobInfo) getFury().deserialize(event.getPrevKV().getValue().getBytes());
                                executorService.execute(new JobExecThread(jobInfo));
                                executorService.execute(new JobPushThread(jobInfo));
                            }
                        }
                    }

                    @Override
                    public void onError(Throwable throwable) {

                    }

                    @Override
                    public void onCompleted() {

                    }
                });
            }
        });
        this.latch.start();
    }

    /***
     * 销毁线程池回收资源
     * @throws Exception
     */
    @Override
    public void destroy() throws Exception {
        this.executorService.shutdownNow();
        this.latch.stop();
    }

    //开启线程执行定时任务
    class JobExecThread implements Runnable {

        private JobInfo jobInfo;

        public JobExecThread(JobInfo jobInfo) {
            this.jobInfo = jobInfo;
        }

        @Override
        public void run() {
            try {
                Class clazz = Class.forName(jobInfo.getJobClass());
                Job job = (Job) clazz.newInstance();
                job.execute(jobInfo);
            } catch (ClassNotFoundException e) {
                logger.warn("加载任务类失败: " + e.getClass());
            } catch (InstantiationException | IllegalAccessException e) {
                logger.warn("执行任务失败: " + e.getCause());
            }
        }
    }

    //开启线程把周期性任务push到列表
    class JobPushThread implements Runnable {

        private JobInfo jobInfo;

        public JobPushThread(JobInfo jobInfo) {
            this.jobInfo = jobInfo;
        }

        @Override
        public void run() {
            if (jobInfo != null && jobInfo.isRepeated()) {
                try {
                    ByteSequence jobKey = ByteSequence.from(JOB_RUN + jobInfo.getJobId(), StandardCharsets.UTF_8);
                    getKVClient().put(jobKey, ByteSequence.from(getFury().serialize(jobInfo)), PutOption.newBuilder().withLeaseId(getLeaseClient().grant(jobInfo.getInterval()).get().getID()).build());
                } catch (Exception e) {
                    logger.error("添加定时任务失败", e);
                }
            }
        }
    }

    //任务初始化线程， 把保存区的任务刷入到运行区
    class JobInitThread implements Runnable {

        @Override
        public void run() {
            try {
                List<KeyValue> jobList = etcdService.getChildList(JOB_LIST);
                for (KeyValue keyValue : jobList) {
                    JobInfo jobInfo = (JobInfo) getFury().deserialize(keyValue.getValue().getBytes());
                    if (jobInfo != null && jobInfo.isRepeated()) {
                        if (!etcdService.isExist(JOB_RUN + jobInfo.getJobId())) {
                            Class clazz = Class.forName(jobInfo.getJobClass());
                            createAndSetRepeatJobInTx(jobInfo.getJobId(), clazz, jobInfo.getInterval());
                        }
                    }
                }
            } catch (Exception e) {
                logger.warn("初始化任务失败: " + e.getMessage());
            }
        }
    }
}
